#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Project Name: cooperation_proj
Version: 
File:    
Author: RoyceCAI
Created Date: 2025-11-03
Description: 
"""

"""
Redis安全审计规则定义
用于检查Redis数据库的安全配置
"""

import subprocess
import re
import os
import configparser
from typing import Dict, List, Any
import logging

logger = logging.getLogger(__name__)


class RedisAuditRules:
    """Redis安全审计规则类"""

    # BASE_COMMAND = f"redis-cli -h localhost -p 6378 -a 666888"   # Redis连接命令
    HOST = "localhost"
    PORT = 6378
    PASSWORD = "666888"

    @staticmethod
    def run_redis_command(command: str, host: str = None, port: int = None, password: str = None) -> str:
        """执行Redis命令"""
        if host is None:
            host = RedisAuditRules.HOST
        if port is None:
            port = RedisAuditRules.PORT
        if password is None:
            password = RedisAuditRules.PASSWORD

        try:
            # 构建Redis连接参数
            auth_args = ""
            if password:
                auth_args = f"-a {password}"

            # 执行Redis命令
            cmd = f"redis-cli {auth_args} -h {host} -p {port} {command}"
            result = subprocess.run(
                cmd,
                shell=True,
                capture_output=True,
                timeout=30,
                text=True
            )
            return result.stdout.strip() if result.stdout else result.stderr.strip()
        except Exception as e:
            return f"Error: {str(e)}"

    @staticmethod
    def get_redis_config_file() -> str:
        """获取Redis配置文件路径"""
        possible_paths = [
            "/etc/redis/redis.conf",
            "/etc/redis.conf",
            "/usr/local/etc/redis.conf",
            "/opt/redis/redis.conf",
            "C:\\Program Files\\Redis\\redis.conf",
            "C:\\Redis\\redis.conf",
            "./redis.conf"
        ]

        for path in possible_paths:
            expanded_path = os.path.expanduser(path)
            if os.path.exists(expanded_path):
                return expanded_path

        # 尝试从redis-cli获取配置路径
        try:
            result = subprocess.run(
                "redis-cli config get dir",
                shell=True,
                capture_output=True,
                text=True
            )
            if "dir" in result.stdout:
                lines = result.stdout.strip().split('\n')
                if len(lines) >= 2:
                    redis_dir = lines[1]
                    conf_path = os.path.join(redis_dir, "../redis.conf")
                    if os.path.exists(conf_path):
                        return conf_path
        except:
            pass

        return ""

    @staticmethod
    def check_authentication() -> Dict[str, Any]:
        """1. 检查认证配置"""
        print("[规则1] 检查Redis认证配置...")
        status_details = []
        issues = []

        # 尝试无密码连接
        test_output = RedisAuditRules.run_redis_command("ping")

        if "NOAUTH" in test_output or "Authentication required" in test_output:
            status_details.append("✓ 认证要求: 已启用")

            # 检查配置文件中的requirepass
            config_file = RedisAuditRules.get_redis_config_file()
            if config_file:
                try:
                    with open(config_file, 'r') as f:
                        content = f.read()
                        requirepass_match = re.search(r"requirepass\s+(.+)", content)
                        if requirepass_match:
                            password_set = requirepass_match.group(1).strip()
                            if password_set and password_set != '""' and password_set != "''":
                                status_details.append("✓ 密码配置: 在配置文件中已设置")
                            else:
                                issues.append("密码配置为空或无效")
                                status_details.append("✗✗ 密码配置: 为空或无效")
                except:
                    status_details.append("⚠️ 无法读取配置文件")
        else:
            if "PONG" in test_output:
                issues.append("未启用认证")
                status_details.append("✗✗ 认证要求: 未启用(任意用户可访问)")
            else:
                status_details.append("⚠️ 无法连接Redis服务")

        return {
            "rule_name": "Redis认证检查",
            "status": "通过" if not issues and "✓ 认证要求: 已启用" in status_details else "未通过",
            "config_status": {
                "authentication_enabled": "NOAUTH" in test_output or "Authentication required" in test_output,
                "config_file_found": bool(config_file)
            },
            "status_details": status_details,
            "recommendation": "认证配置安全" if not issues else "建议设置requirepass启用认证"
        }

    @staticmethod
    def check_network_security() -> Dict[str, Any]:
        """2. 检查网络安全配置"""
        print("[规则2] 检查Redis网络安全...")
        status_details = []
        issues = []

        # 检查绑定地址
        config_command = f"config get bind"
        bind_output = RedisAuditRules.run_redis_command(config_command)
        bind_match = re.search(r"bind\s+([^\n]+)", bind_output)
        bind_address = bind_match.group(1).strip() if bind_match else ""

        if bind_address == "127.0.0.1":
            status_details.append("✓ 绑定地址: 127.0.0.1(仅本地访问)")
        elif not bind_address or bind_address == "":
            issues.append("绑定地址为空(监听所有接口)")
            status_details.append("✗✗ 绑定地址: 空(监听所有网络接口)")
        else:
            status_details.append(f"⚠️ 绑定地址: {bind_address}")

        # 检查保护模式
        protected_mode_output = RedisAuditRules.run_redis_command("config get protected-mode")
        protected_match = re.search(r"protected-mode\s+(\w+)", protected_mode_output)
        protected_mode = protected_match.group(1) if protected_match else ""

        if protected_mode == "yes":
            status_details.append("✓ 保护模式: 已启用")
        else:
            issues.append("保护模式未启用")
            status_details.append("✗✗ 保护模式: 未启用")

        # 检查端口配置
        port_output = RedisAuditRules.run_redis_command("config get port")
        port_match = re.search(r"port\s+(\d+)", port_output)
        port = port_match.group(1) if port_match else "6379"

        if port == "6379":
            status_details.append("⚠️ 使用默认端口: 6379(建议修改)")
        else:
            status_details.append(f"✓ 使用非默认端口: {port}")

        return {
            "rule_name": "Redis网络安全检查",
            "status": "正常" if bind_address == "127.0.0.1" and protected_mode == "yes" else "异常",
            "config_status": {
                "bind_address": bind_address,
                "protected_mode": protected_mode,
                "port": port
            },
            "status_details": status_details,
            "recommendation": "网络配置安全" if bind_address == "127.0.0.1" else "建议绑定到127.0.0.1并启用保护模式"
        }

    @staticmethod
    def check_dangerous_commands() -> Dict[str, Any]:
        """3. 检查危险命令配置"""
        print("[规则3] 检查Redis危险命令...")
        status_details = []
        issues = []

        # 检查危险命令的重命名配置
        dangerous_commands = {
            "FLUSHALL": "清空所有数据",
            "FLUSHDB": "清空当前数据库",
            "CONFIG": "修改配置",
            "KEYS": "枚举所有键(可能影响性能)",
            "SHUTDOWN": "关闭服务",
            "DEBUG": "调试命令",
            "SAVE": "保存数据",
            "BGSAVE": "后台保存数据"
        }

        disabled_commands = []
        for cmd in dangerous_commands.keys():
            rename_output = RedisAuditRules.run_redis_command(f"config get rename-command {cmd}")
            if "rename-command" in rename_output:
                match = re.search(rf"rename-command {cmd}\s+(\S+)", rename_output)
                if match:
                    new_name = match.group(1)
                    if new_name == '""':
                        status_details.append(f"✓ 危险命令 {cmd}: 已禁用")
                        disabled_commands.append(cmd)
                    else:
                        status_details.append(f"⚠️ 危险命令 {cmd}: 重命名为 {new_name}")
                else:
                    status_details.append(f"✗✗ 危险命令 {cmd}: 未重命名或禁用")
                    issues.append(f"{cmd}命令未限制")
            else:
                status_details.append(f"✗✗ 危险命令 {cmd}: 未重命名或禁用")
                issues.append(f"{cmd}命令未限制")

        return {
            "rule_name": "Redis危险命令检查",
            "status": "正常" if len(disabled_commands) >= 4 else "异常",  # 至少禁用4个关键命令
            "config_status": {
                "disabled_commands_count": len(disabled_commands),
                "disabled_commands": disabled_commands
            },
            "status_details": status_details,
            "recommendation": f"已禁用{len(disabled_commands)}个危险命令" if disabled_commands else "建议重命名或禁用FLUSHALL、FLUSHDB、CONFIG、KEYS等危险命令"
        }

    @staticmethod
    def check_acl_configuration() -> Dict[str, Any]:
        """4. 检查ACL配置(Redis 6.0+)"""
        print("[规则4] 检查Redis ACL配置...")
        status_details = []
        issues = []

        # 检查ACL功能是否启用
        acl_output = RedisAuditRules.run_redis_command("acl list")

        if "NOAUTH" in acl_output or "Authentication required" in acl_output:
            # 需要认证，尝试使用默认用户
            status_details.append("✓ 认证已启用")

            # 检查是否有多个用户配置
            config_file = RedisAuditRules.get_redis_config_file()
            if config_file:
                try:
                    with open(config_file, 'r') as f:
                        content = f.read()
                        acl_pattern = re.findall(r"user\s+\w+\s+on", content)
                        if len(acl_pattern) > 1:
                            status_details.append(f"✓ 多用户配置: 发现{len(acl_pattern)}个用户")
                        else:
                            status_details.append("⚠️ 多用户配置: 仅默认用户")
                except:
                    status_details.append("⚠️ 无法读取ACL配置")
        elif "command not found" in acl_output or "unknown command" in acl_output:
            status_details.append("ℹ️ ACL功能: Redis版本可能低于6.0")
        else:
            # 解析ACL列表
            acl_lines = acl_output.split('\n')
            user_count = len([line for line in acl_lines if line.startswith('user')])
            status_details.append(f"✓ ACL用户数量: {user_count}")

        return {
            "rule_name": "Redis ACL检查",
            "status": "正常",
            "config_status": {
                "acl_supported": "command not found" not in acl_output,
                "user_count": user_count if 'user_count' in locals() else 0
            },
            "status_details": status_details,
            "recommendation": "建议使用Redis 6.0+的ACL功能进行精细权限控制"
        }

    @staticmethod
    def check_persistence_config() -> Dict[str, Any]:
        """5. 检查持久化配置"""
        print("[规则5] 检查Redis持久化配置...")
        status_details = []
        issues = []

        # 检查RDB配置
        save_output = RedisAuditRules.run_redis_command("config get save")
        save_match = re.findall(r"save\s+([^\n]+)", save_output)

        if save_match:
            status_details.append("✓ RDB持久化: 已配置")
            for save_config in save_match:
                status_details.append(f"  - {save_config}")
        else:
            issues.append("RDB持久化未配置")
            status_details.append("✗✗ RDB持久化: 未配置")

        # 检查AOF配置
        aof_output = RedisAuditRules.run_redis_command("config get appendonly")
        aof_match = re.search(r"appendonly\s+(\w+)", aof_output)
        aof_enabled = aof_match.group(1) if aof_match else "no"

        if aof_enabled == "yes":
            status_details.append("✓ AOF持久化: 已启用")

            # 检查AOF策略
            aof_fsync_output = RedisAuditRules.run_redis_command("config get appendfsync")
            aof_fsync_match = re.search(r"appendfsync\s+(\w+)", aof_fsync_output)
            aof_fsync = aof_fsync_match.group(1) if aof_fsync_match else ""
            status_details.append(f"  - AOF同步策略: {aof_fsync}")
        else:
            status_details.append("⚠️ AOF持久化: 未启用")

        return {
            "rule_name": "Redis持久化检查",
            "status": "通过" if save_match else "未通过",
            "config_status": {
                "rdb_enabled": bool(save_match),
                "aof_enabled": aof_enabled == "yes",
                "aof_fsync": aof_fsync if 'aof_fsync' in locals() else ""
            },
            "status_details": status_details,
            "recommendation": "持久化配置合理" if save_match else "建议配置RDB或AOF持久化防止数据丢失"
        }

    @staticmethod
    def check_memory_config() -> Dict[str, Any]:
        """6. 检查内存配置"""
        print("[规则6] 检查Redis内存配置...")
        status_details = []
        issues = []

        # 检查内存使用情况
        info_memory = RedisAuditRules.run_redis_command("info memory")
        used_memory_match = re.search(r"used_memory:(\d+)", info_memory)
        max_memory_match = re.search(r"maxmemory:(\d+)", info_memory)
        maxmemory_policy_match = re.search(r"maxmemory_policy:(\w+)", info_memory)

        if used_memory_match:
            used_memory = int(used_memory_match.group(1))
            used_memory_human = used_memory // (1024 * 1024)  # 转换为MB
            status_details.append(f"✓ 已用内存: {used_memory_human}MB")

        if max_memory_match:
            max_memory = int(max_memory_match.group(1))
            if max_memory == 0:
                issues.append("未设置内存限制")
                status_details.append("✗✗ 最大内存: 未设置(可能内存溢出)")
            else:
                max_memory_human = max_memory // (1024 * 1024)
                status_details.append(f"✓ 最大内存: {max_memory_human}MB")

        if maxmemory_policy_match:
            policy = maxmemory_policy_match.group(1)
            policies = {
                "noeviction": "不淘汰(推荐)",
                "allkeys-lru": "全体LRU淘汰",
                "volatile-lru": "过期键LRU淘汰"
            }
            policy_desc = policies.get(policy, policy)
            status_details.append(f"✓ 内存淘汰策略: {policy_desc}")

        # 检查内存碎片率
        mem_fragmentation_ratio_match = re.search(r"mem_fragmentation_ratio:([\d.]+)", info_memory)
        if mem_fragmentation_ratio_match:
            ratio = float(mem_fragmentation_ratio_match.group(1))
            if ratio > 1.5:
                issues.append(f"内存碎片率过高: {ratio}")
                status_details.append(f"✗✗ 内存碎片率: {ratio}(建议<1.5)")
            else:
                status_details.append(f"✓ 内存碎片率: {ratio}")

        return {
            "rule_name": "Redis内存配置检查",
            "status": "正常" if max_memory_match and max_memory_match.group(1) != "0" else "异常",
            "config_status": {
                "maxmemory_set": max_memory_match and max_memory_match.group(1) != "0",
                "memory_policy": maxmemory_policy_match.group(1) if maxmemory_policy_match else "",
                "fragmentation_ratio": float(
                    mem_fragmentation_ratio_match.group(1)) if mem_fragmentation_ratio_match else 0
            },
            "status_details": status_details,
            "recommendation": "内存配置合理" if max_memory_match and max_memory_match.group(
                1) != "0" else "建议设置maxmemory限制"
        }

    @staticmethod
    def check_logging_config() -> Dict[str, Any]:
        """7. 检查日志配置"""
        print("[规则7] 检查Redis日志配置...")
        status_details = []
        issues = []

        # 检查日志级别
        loglevel_output = RedisAuditRules.run_redis_command("config get loglevel")
        loglevel_match = re.search(r"loglevel\s+(\w+)", loglevel_output)
        loglevel = loglevel_match.group(1) if loglevel_match else "notice"

        loglevels = {"debug": "调试", "verbose": "详细", "notice": "通知", "warning": "警告"}
        level_desc = loglevels.get(loglevel, loglevel)
        status_details.append(f"✓ 日志级别: {level_desc}")

        if loglevel == "debug":
            status_details.append("⚠️ 日志级别为debug(可能产生大量日志)")

        # 检查日志文件配置
        logfile_output = RedisAuditRules.run_redis_command("config get logfile")
        logfile_match = re.search(r"logfile\s+(\S+)", logfile_output)
        logfile = logfile_match.group(1) if logfile_match else ""

        if logfile and logfile != '""':
            status_details.append(f"✓ 日志文件: {logfile}")
        else:
            issues.append("未配置日志文件")
            status_details.append("✗✗ 日志文件: 未配置(输出到标准输出)")

        # 检查慢查询日志
        slowlog_output = RedisAuditRules.run_redis_command("config get slowlog-log-slower-than")
        slowlog_match = re.search(r"slowlog-log-slower-than\s+(\d+)", slowlog_output)
        slowlog_threshold = int(slowlog_match.group(1)) if slowlog_match else 10000

        status_details.append(f"✓ 慢查询阈值: {slowlog_threshold}微秒")

        return {
            "rule_name": "Redis日志配置检查",
            "status": "正常" if logfile and logfile != '""' else "异常",
            "config_status": {
                "loglevel": loglevel,
                "logfile_configured": bool(logfile and logfile != '""'),
                "slowlog_threshold": slowlog_threshold
            },
            "status_details": status_details,
            "recommendation": "日志配置合理" if logfile and logfile != '""' else "建议配置logfile指定日志文件路径"
        }

    @staticmethod
    def check_connection_security() -> Dict[str, Any]:
        """8. 检查连接安全"""
        print("[规则8] 检查Redis连接安全...")
        status_details = []
        issues = []

        # 检查最大客户端连接数
        maxclients_output = RedisAuditRules.run_redis_command("config get maxclients")
        maxclients_match = re.search(r"maxclients\s+(\d+)", maxclients_output)
        maxclients = int(maxclients_match.group(1)) if maxclients_match else 10000

        status_details.append(f"✓ 最大客户端数: {maxclients}")

        if maxclients > 10000:
            status_details.append("⚠️ 最大客户端数设置较高")

        # 检查超时配置
        timeout_output = RedisAuditRules.run_redis_command("config get timeout")
        timeout_match = re.search(r"timeout\s+(\d+)", timeout_output)
        timeout = int(timeout_match.group(1)) if timeout_match else 0

        if timeout > 0:
            status_details.append(f"✓ 客户端超时: {timeout}秒")
        else:
            issues.append("未设置客户端超时")
            status_details.append("✗✗ 客户端超时: 未设置")

        # 检查TCP-Keepalive
        tcp_keepalive_output = RedisAuditRules.run_redis_command("config get tcp-keepalive")
        tcp_keepalive_match = re.search(r"tcp-keepalive\s+(\d+)", tcp_keepalive_output)
        tcp_keepalive = int(tcp_keepalive_match.group(1)) if tcp_keepalive_match else 300

        status_details.append(f"✓ TCP保活: {tcp_keepalive}秒")

        return {
            "rule_name": "Redis连接安全检查",
            "status": "正常" if timeout > 0 else "异常",
            "config_status": {
                "maxclients": maxclients,
                "timeout_set": timeout > 0,
                "tcp_keepalive": tcp_keepalive
            },
            "status_details": status_details,
            "recommendation": "连接配置合理" if timeout > 0 else "建议设置timeout防止空闲连接占用资源"
        }

    @staticmethod
    def get_all_rules() -> List[callable]:
        """获取所有审计规则的函数列表"""
        return [
            RedisAuditRules.check_authentication,
            RedisAuditRules.check_network_security,
            RedisAuditRules.check_dangerous_commands,
            RedisAuditRules.check_acl_configuration,
            RedisAuditRules.check_persistence_config,
            RedisAuditRules.check_memory_config,
            RedisAuditRules.check_logging_config,
            RedisAuditRules.check_connection_security
        ]
